package com.lagou.edu.zk;

import com.lagou.edu.datasource.ZkDataSourceFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.util.StringUtils;

/**
 * ZK工具类
 */
@Slf4j
public class ZkHelper {
    // 数据库信息节点
    public static final String DB_NODE_PATH = "/db";
    // 默认连接信息
    public static final String[] defaultConnection = new String[]{"127.0.0.1:3306", "root", "123456"};
    private static CuratorFramework client = null;

    /**
     * 获取ZK客户端
     *
     * @return
     * @throws Exception
     */
    public static CuratorFramework getClient() throws Exception {
        //客户端不存在时，初始化连接
        if (client == null) {
            initClient();
        }
        return client;
    }

    /**
     * 初始化数据库信息节点
     *
     * @throws Exception
     */
    public static void initDbInfoNode() throws Exception {
        String data = StringUtils.arrayToDelimitedString(defaultConnection, ",");
        getClient().create().orSetData().forPath(ZkHelper.DB_NODE_PATH, data.getBytes());
        log.info("创建/更新 数据库监听节点[{}]，设置默认数据库连接 {}", DB_NODE_PATH, data);
    }

    /**
     * 初始化ZK客户端，通过虚拟机同步方法防止重复初始化
     *
     * @throws Exception
     */
    private synchronized static void initClient() throws Exception {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory
                .builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(10000)
                .retryPolicy(retry)
                .namespace("task3")
                .build();
        client.start();
        // 客户端单例
        ZkHelper.client = client;
        // 检测服务注册节点，不存在则创建
        initDbInfoNode();
    }

    /**
     * 监听数据库信息变化
     *
     * @throws Exception
     */
    public static void listen() throws Exception {
        CuratorFramework client = ZkHelper.getClient();
        // 设置监听路径
        NodeCache nodeCache = new NodeCache(client, DB_NODE_PATH);
        // 设置监听方法
        nodeCache.getListenable().addListener(() -> {
            String data = new String(client.getData().forPath(DB_NODE_PATH));
            log.info("ZooKeeper >> 数据库连接更新 {}", data);
            String[] dataParts = data.split(",");
            // 数据库信息有更新时，重新配置数据源
            ZkDataSourceFactory.config(dataParts[0], dataParts[1], dataParts[2]);

        });
        nodeCache.start();
        log.info("监听Zookeeper数据库信息节点 {}", DB_NODE_PATH);
        log.info("Zookeeper数据库信息节点格式 addr,username,password");
    }
}
